//! # 内存消息总线实现
//!
//! 提供基于内存的高性能消息总线实现，适用于单进程内的消息传递

use async_trait::async_trait;
use dashmap::DashMap;
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::sync::mpsc;
use uuid::Uuid;

use crate::bus::{MessageBus, MessageBusStats, Subscriber};
use crate::config::MessageBusConfig;
use crate::error::{MessageBusError, MessageBusResult};
use crate::message::Message;
use crate::router::{MessageRouter, TopicPattern};
use crate::subscriber::{EnhancedMemorySubscriber, SubscriberConfig};

/// 订阅者信息
#[derive(Debug)]
struct SubscriberInfo {
    /// 订阅模式
    pattern: String,
    /// 消息发送器
    sender: mpsc::Sender<Message>,
    /// 创建时间
    created_at: chrono::DateTime<chrono::Utc>,
    /// 是否活跃
    is_active: Arc<std::sync::atomic::AtomicBool>,
}

/// 内存消息总线实现
pub struct InMemoryMessageBus {
    /// 配置
    config: MessageBusConfig,
    /// 订阅者映射（订阅ID -> 订阅者信息）
    subscribers: Arc<DashMap<String, SubscriberInfo>>,
    /// 消息路由器
    router: Arc<parking_lot::RwLock<MessageRouter>>,
    /// 统计信息
    stats: Arc<MessageBusStatsInternal>,
    /// 是否已关闭
    is_closed: Arc<std::sync::atomic::AtomicBool>,
    /// 启动时间
    started_at: chrono::DateTime<chrono::Utc>,
}

/// 内部统计信息结构
#[derive(Debug)]
struct MessageBusStatsInternal {
    total_published: AtomicU64,
    total_received: AtomicU64,
    total_failed: AtomicU64,
    active_subscribers: AtomicUsize,
    active_topics: AtomicUsize,
    queue_length: AtomicUsize,
    total_message_size: AtomicU64,
    message_count_for_size: AtomicU64,
}

impl Default for MessageBusStatsInternal {
    fn default() -> Self {
        Self {
            total_published: AtomicU64::new(0),
            total_received: AtomicU64::new(0),
            total_failed: AtomicU64::new(0),
            active_subscribers: AtomicUsize::new(0),
            active_topics: AtomicUsize::new(0),
            queue_length: AtomicUsize::new(0),
            total_message_size: AtomicU64::new(0),
            message_count_for_size: AtomicU64::new(0),
        }
    }
}

impl InMemoryMessageBus {
    /// 创建新的内存消息总线
    pub async fn new() -> MessageBusResult<Self> {
        tracing::info!("正在创建新的内存消息总线实例");
        let bus = Self::builder().build().await?;
        tracing::info!("内存消息总线实例创建成功");
        Ok(bus)
    }

    /// 创建建造者
    pub fn builder() -> InMemoryMessageBusBuilder {
        InMemoryMessageBusBuilder::new()
    }

    /// 使用指定配置创建内存消息总线
    pub async fn with_config(config: MessageBusConfig) -> MessageBusResult<Self> {
        tracing::info!(
            "正在创建内存消息总线，配置: 缓冲区大小={}, 最大消息大小={} bytes, 最大连接数={}",
            config.buffer_size,
            config.max_message_size,
            config.max_connections
        );
        
        config.validate()?;
        tracing::debug!("消息总线配置验证通过");

        let bus = Self {
            config,
            subscribers: Arc::new(DashMap::new()),
            router: Arc::new(parking_lot::RwLock::new(MessageRouter::new())),
            stats: Arc::new(MessageBusStatsInternal::default()),
            is_closed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            started_at: chrono::Utc::now(),
        };

        // 启动后台清理任务
        tracing::debug!("启动内存消息总线清理任务");
        bus.start_cleanup_task().await;
        tracing::debug!("内存消息总线创建完成");

        Ok(bus)
    }

    /// 启动后台清理任务
    async fn start_cleanup_task(&self) {
        let subscribers = self.subscribers.clone();
        let stats = self.stats.clone();
        let is_closed = self.is_closed.clone();

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(30));

            while !is_closed.load(Ordering::Relaxed) {
                interval.tick().await;

                // 清理不活跃的订阅者
                let mut inactive_subscribers = Vec::new();
                for entry in subscribers.iter() {
                    if !entry.value().is_active.load(Ordering::Relaxed) 
                        || entry.value().sender.is_closed() {
                        inactive_subscribers.push(entry.key().clone());
                    }
                }

                for subscriber_id in inactive_subscribers {
                    if let Some((_, _)) = subscribers.remove(&subscriber_id) {
                        stats.active_subscribers.fetch_sub(1, Ordering::Relaxed);
                        tracing::debug!("清理不活跃订阅者: {}", subscriber_id);
                    }
                }
            }
        });
    }

    /// 检查主题是否匹配模式
    fn matches_pattern(pattern: &str, topic: &str) -> bool {
        let topic_pattern = if pattern.contains('*') {
            TopicPattern::wildcard(pattern)
        } else {
            TopicPattern::exact(pattern)
        };
        topic_pattern.matches(topic)
    }

    /// 获取匹配的订阅者
    fn get_matching_subscribers(&self, topic: &str) -> Vec<String> {
        let mut matching_subscribers = Vec::new();

        for entry in self.subscribers.iter() {
            let subscriber_info = entry.value();
            if subscriber_info.is_active.load(Ordering::Relaxed) 
                && !subscriber_info.sender.is_closed()
                && Self::matches_pattern(&subscriber_info.pattern, topic) {
                matching_subscribers.push(entry.key().clone());
            }
        }

        matching_subscribers
    }

    /// 发送消息给指定订阅者
    async fn send_to_subscriber(&self, subscriber_id: &str, message: Message) -> MessageBusResult<()> {
        if let Some(subscriber_info) = self.subscribers.get(subscriber_id) {
            match subscriber_info.sender.send(message).await {
                Ok(()) => {
                    self.stats.total_received.fetch_add(1, Ordering::Relaxed);
                    Ok(())
                }
                Err(_) => {
                    // 发送失败，可能是通道已关闭，标记为不活跃
                    subscriber_info.is_active.store(false, Ordering::Relaxed);
                    Err(MessageBusError::publish_error(subscriber_id, "发送到订阅者失败"))
                }
            }
        } else {
            Err(MessageBusError::publish_error(subscriber_id, "订阅者不存在"))
        }
    }

    /// 更新统计信息
    fn update_stats(&self, message: &Message) {
        self.stats.total_published.fetch_add(1, Ordering::Relaxed);
        
        let message_size = message.size();
        self.stats.total_message_size.fetch_add(message_size as u64, Ordering::Relaxed);
        self.stats.message_count_for_size.fetch_add(1, Ordering::Relaxed);

        // 更新活跃主题数（简化实现，实际应该跟踪唯一主题）
        self.stats.active_topics.store(
            self.subscribers.len().min(1000), // 限制最大值防止溢出
            Ordering::Relaxed
        );
    }
}

#[async_trait]
impl MessageBus for InMemoryMessageBus {
    async fn publish(&self, mut message: Message) -> MessageBusResult<()> {
        let start_time = std::time::Instant::now();
        let topic = message.topic().to_string();
        
        tracing::debug!(
            "开始发布消息: topic={}, message_id={}, size={} bytes",
            topic,
            message.id(),
            message.size()
        );
        
        if self.is_closed.load(Ordering::Relaxed) {
            tracing::error!("尝试在已关闭的消息总线上发布消息: {}", topic);
            return Err(MessageBusError::publish_error(
                message.topic(),
                "消息总线已关闭",
            ));
        }

        // 验证消息
        if let Err(e) = message.validate() {
            tracing::error!("消息验证失败: topic={}, error={}", topic, e);
            return Err(e);
        }
        tracing::trace!("消息验证通过: {}", topic);

        // 检查消息大小
        if message.size() > self.config.max_message_size {
            let error_msg = format!("消息大小 {} 超过最大限制 {}", message.size(), self.config.max_message_size);
            tracing::error!("消息大小超限: topic={}, {}", topic, error_msg);
            return Err(MessageBusError::message_format_error(error_msg));
        }

        // 标记消息为已发送
        message.metadata_mut().mark_sent();
        tracing::trace!("消息已标记为已发送: {}", topic);

        // 更新统计信息
        self.update_stats(&message);

        // 获取匹配的订阅者
        let matching_subscribers = self.get_matching_subscribers(message.topic());
        let subscriber_count = matching_subscribers.len();
        
        tracing::debug!("找到 {} 个匹配的订阅者: topic={}", subscriber_count, topic);

        if matching_subscribers.is_empty() {
            tracing::debug!("没有找到匹配的订阅者: {}", message.topic());
            let duration = start_time.elapsed();
            tracing::trace!("消息发布完成 (无订阅者): topic={}, duration={:?}", topic, duration);
            return Ok(());
        }

        // 向所有匹配的订阅者发送消息
        let mut send_tasks = Vec::new();
        for subscriber_id in matching_subscribers {
            let message_clone = message.clone();
            let subscriber_id_clone = subscriber_id.clone();
            
            if let Some(subscriber_info) = self.subscribers.get(&subscriber_id) {
                let sender = subscriber_info.sender.clone();
                let task = tokio::spawn(async move {
                    match sender.send(message_clone).await {
                        Ok(()) => Ok(()),
                        Err(_) => Err(MessageBusError::publish_error(
                            &subscriber_id_clone,
                            "发送到订阅者失败",
                        )),
                    }
                });
                send_tasks.push(task);
            }
        }

        // 等待所有发送任务完成
        let mut failed_count = 0;
        let mut success_count = 0;
        for task in send_tasks {
            match task.await {
                Ok(Ok(())) => {
                    self.stats.total_received.fetch_add(1, Ordering::Relaxed);
                    success_count += 1;
                }
                Ok(Err(_)) | Err(_) => {
                    failed_count += 1;
                    self.stats.total_failed.fetch_add(1, Ordering::Relaxed);
                }
            }
        }

        let duration = start_time.elapsed();
        
        if failed_count > 0 {
            tracing::warn!(
                "消息发布部分失败: topic={}, 成功={}, 失败={}, duration={:?}",
                topic, success_count, failed_count, duration
            );
        } else {
            tracing::info!(
                "消息发布成功: topic={}, subscribers={}, duration={:?}", 
                topic, success_count, duration
            );
        }

        Ok(())
    }

    async fn subscribe(&self, pattern: &str) -> MessageBusResult<Box<dyn Subscriber>> {
        tracing::debug!("开始创建订阅: pattern={}", pattern);
        
        if self.is_closed.load(Ordering::Relaxed) {
            tracing::error!("尝试在已关闭的消息总线上创建订阅: pattern={}", pattern);
            return Err(MessageBusError::subscribe_error(
                pattern,
                "消息总线已关闭",
            ));
        }

        // 检查订阅者数量限制
        let current_subscribers = self.stats.active_subscribers.load(Ordering::Relaxed);
        if current_subscribers as usize >= self.config.max_connections {
            tracing::warn!(
                "订阅者数量超限: 当前={}, 最大={}, pattern={}",
                current_subscribers, 
                self.config.max_connections, 
                pattern
            );
            return Err(MessageBusError::subscribe_error(
                pattern,
                "订阅者数量超过限制",
            ));
        }

        let subscription_id = format!("sub-{}", Uuid::new_v4());
        tracing::trace!("生成订阅ID: {}", subscription_id);
        
        let (sender, receiver) = mpsc::channel(self.config.buffer_size);

        let subscriber_config = SubscriberConfig {
            buffer_size: self.config.buffer_size,
            auto_ack: true,
            max_unacked: 100,
            message_timeout_ms: 30000,
            prefetch_count: 10,
        };
        
        tracing::trace!(
            "创建订阅者配置: buffer_size={}, auto_ack={}, max_unacked={}",
            subscriber_config.buffer_size,
            subscriber_config.auto_ack,
            subscriber_config.max_unacked
        );

        let subscriber = EnhancedMemorySubscriber::new(
            subscription_id.clone(),
            pattern.to_string(),
            subscriber_config,
            receiver,
        );

        let subscriber_info = SubscriberInfo {
            pattern: pattern.to_string(),
            sender,
            created_at: chrono::Utc::now(),
            is_active: subscriber.is_active.clone(),
        };

        self.subscribers.insert(subscription_id.clone(), subscriber_info);
        self.stats.active_subscribers.fetch_add(1, Ordering::Relaxed);

        tracing::info!(
            "订阅创建成功: id={}, pattern={}, 当前订阅者数={}", 
            subscription_id,
            pattern,
            self.stats.active_subscribers.load(Ordering::Relaxed)
        );

        Ok(Box::new(subscriber))
    }

    async fn unsubscribe(&self, subscription_id: &str) -> MessageBusResult<()> {
        tracing::debug!("尝试取消订阅: id={}", subscription_id);
        
        if let Some((_, subscriber_info)) = self.subscribers.remove(subscription_id) {
            subscriber_info.is_active.store(false, Ordering::Relaxed);
            let current_count = self.stats.active_subscribers.fetch_sub(1, Ordering::Relaxed) - 1;
            
            tracing::info!(
                "订阅取消成功: id={}, pattern={}, 当前订阅者数={}",
                subscription_id,
                subscriber_info.pattern,
                current_count
            );
            Ok(())
        } else {
            tracing::warn!("尝试取消不存在的订阅: id={}", subscription_id);
            Err(MessageBusError::subscribe_error(
                subscription_id,
                "订阅不存在",
            ))
        }
    }

    async fn subscriber_count(&self, topic: &str) -> MessageBusResult<usize> {
        let count = self.get_matching_subscribers(topic).len();
        Ok(count)
    }

    async fn list_topics(&self) -> MessageBusResult<Vec<String>> {
        let mut topics = std::collections::HashSet::new();

        for entry in self.subscribers.iter() {
            let pattern = &entry.value().pattern;
            // 对于精确匹配的模式，直接添加
            if !pattern.contains('*') {
                topics.insert(pattern.clone());
            }
        }

        Ok(topics.into_iter().collect())
    }

    async fn has_subscribers(&self, topic: &str) -> MessageBusResult<bool> {
        let count = self.subscriber_count(topic).await?;
        Ok(count > 0)
    }

    async fn close(&self) -> MessageBusResult<()> {
        if self.is_closed.swap(true, Ordering::Relaxed) {
            return Ok(()); // 已经关闭
        }

        tracing::info!("正在关闭内存消息总线...");

        // 关闭所有订阅者
        for entry in self.subscribers.iter() {
            entry.value().is_active.store(false, Ordering::Relaxed);
        }

        // 清空订阅者
        self.subscribers.clear();
        self.stats.active_subscribers.store(0, Ordering::Relaxed);

        tracing::info!("内存消息总线已关闭");
        Ok(())
    }

    async fn stats(&self) -> MessageBusResult<MessageBusStats> {
        let uptime = chrono::Utc::now()
            .signed_duration_since(self.started_at)
            .num_seconds() as u64;

        let total_published = self.stats.total_published.load(Ordering::Relaxed);
        let messages_per_second = if uptime > 0 {
            total_published as f64 / uptime as f64
        } else {
            0.0
        };

        let total_size = self.stats.total_message_size.load(Ordering::Relaxed);
        let message_count = self.stats.message_count_for_size.load(Ordering::Relaxed);
        let average_message_size = if message_count > 0 {
            total_size as usize / message_count as usize
        } else {
            0
        };

        Ok(MessageBusStats {
            total_published,
            total_received: self.stats.total_received.load(Ordering::Relaxed),
            total_failed: self.stats.total_failed.load(Ordering::Relaxed),
            active_subscribers: self.stats.active_subscribers.load(Ordering::Relaxed),
            active_topics: self.stats.active_topics.load(Ordering::Relaxed),
            queue_length: self.stats.queue_length.load(Ordering::Relaxed),
            uptime_seconds: uptime,
            messages_per_second,
            average_message_size,
            memory_usage: std::mem::size_of::<Self>(), // 简化实现
        })
    }
}

/// 内存消息总线构建器
pub struct InMemoryMessageBusBuilder {
    config: MessageBusConfig,
}

impl Default for InMemoryMessageBusBuilder {
    fn default() -> Self {
        Self::new()
    }
}

impl InMemoryMessageBusBuilder {
    /// 创建新的构建器
    pub fn new() -> Self {
        let mut config = MessageBusConfig::default();
        config.bus_type = crate::config::MessageBusType::Memory;
        
        Self { config }
    }

    /// 设置缓冲区大小
    pub fn buffer_size(mut self, size: usize) -> Self {
        self.config.buffer_size = size;
        self
    }

    /// 设置最大消息大小
    pub fn max_message_size(mut self, size: usize) -> Self {
        self.config.max_message_size = size;
        self
    }

    /// 设置是否启用指标收集
    pub fn enable_metrics(mut self, enable: bool) -> Self {
        self.config.enable_metrics = enable;
        self
    }

    /// 设置自定义配置
    pub fn with_config(mut self, config: MessageBusConfig) -> Self {
        self.config = config;
        self
    }

    /// 构建消息总线
    pub async fn build(self) -> MessageBusResult<InMemoryMessageBus> {
        InMemoryMessageBus::with_config(self.config).await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::message::Message;
    use serde_json::json;

    #[tokio::test]
    async fn test_memory_message_bus_creation() {
        let bus = InMemoryMessageBus::new().await.unwrap();
        assert!(!bus.is_closed.load(Ordering::Relaxed));
    }

    #[tokio::test]
    async fn test_publish_and_subscribe() {
        let bus = InMemoryMessageBus::new().await.unwrap();

        // 订阅消息
        let subscriber = bus.subscribe("test.events.*").await.unwrap();

        // 发布消息
        let message = Message::new("test.events.login", json!({"user_id": 123}));
        assert!(bus.publish(message).await.is_ok());

        // 接收消息
        let received = subscriber.receive_timeout(1000).await.unwrap();
        assert!(received.is_some());
        assert_eq!(received.unwrap().topic(), "test.events.login");
    }

    #[tokio::test]
    async fn test_pattern_matching() {
        let bus = InMemoryMessageBus::new().await.unwrap();

        // 通配符订阅
        let subscriber1 = bus.subscribe("user.events.*").await.unwrap();
        let subscriber2 = bus.subscribe("user.profile.*").await.unwrap();

        // 发布匹配第一个订阅的消息
        let message = Message::new("user.events.login", json!({"test": "data"}));
        assert!(bus.publish(message).await.is_ok());

        // 第一个订阅者应该收到消息
        let received1 = subscriber1.receive_timeout(100).await.unwrap();
        assert!(received1.is_some());

        // 第二个订阅者不应该收到消息
        let received2 = subscriber2.receive_timeout(100).await.unwrap();
        assert!(received2.is_none());
    }

    #[tokio::test]
    async fn test_multiple_subscribers() {
        let bus = InMemoryMessageBus::new().await.unwrap();

        // 多个订阅者订阅同一模式
        let subscriber1 = bus.subscribe("broadcast.*").await.unwrap();
        let subscriber2 = bus.subscribe("broadcast.*").await.unwrap();

        // 发布消息
        let message = Message::new("broadcast.news", json!({"content": "重要新闻"}));
        assert!(bus.publish(message).await.is_ok());

        // 两个订阅者都应该收到消息
        let received1 = subscriber1.receive_timeout(100).await.unwrap();
        assert!(received1.is_some());

        let received2 = subscriber2.receive_timeout(100).await.unwrap();
        assert!(received2.is_some());
    }

    #[tokio::test]
    async fn test_unsubscribe() {
        let bus = InMemoryMessageBus::new().await.unwrap();

        let subscriber = bus.subscribe("test.topic").await.unwrap();
        let subscription_id = subscriber.subscription_id().to_string();

        // 取消订阅
        assert!(bus.unsubscribe(&subscription_id).await.is_ok());

        // 发布消息
        let message = Message::new("test.topic", json!({"test": "data"}));
        assert!(bus.publish(message).await.is_ok());

        // 不应该收到消息
        let received = subscriber.receive_timeout(100).await.unwrap();
        assert!(received.is_none());
    }

    #[tokio::test]
    async fn test_stats() {
        let bus = InMemoryMessageBus::new().await.unwrap();

        let _subscriber = bus.subscribe("test.*").await.unwrap();

        let message = Message::new("test.topic", json!({"test": "data"}));
        assert!(bus.publish(message).await.is_ok());

        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.total_published, 1);
        assert_eq!(stats.active_subscribers, 1);
    }

    #[tokio::test]
    async fn test_close() {
        let bus = InMemoryMessageBus::new().await.unwrap();

        let _subscriber = bus.subscribe("test.*").await.unwrap();

        // 关闭消息总线
        assert!(bus.close().await.is_ok());

        // 关闭后不能发布消息
        let message = Message::new("test.topic", json!({"test": "data"}));
        assert!(bus.publish(message).await.is_err());

        // 关闭后不能订阅
        assert!(bus.subscribe("new.topic").await.is_err());
    }

    #[tokio::test]
    async fn test_builder() {
        let bus = InMemoryMessageBusBuilder::new()
            .buffer_size(500)
            .max_message_size(2048)
            .enable_metrics(true)
            .build()
            .await
            .unwrap();

        assert_eq!(bus.config.buffer_size, 500);
        assert_eq!(bus.config.max_message_size, 2048);
        assert!(bus.config.enable_metrics);
    }
}